/*! ******************************************************************************
 *
 * Pentaho Data Integration
 *
 * Copyright (C) 2002-2016 by Pentaho : http://www.pentaho.com
 *
 *******************************************************************************
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 ******************************************************************************/

package org.pentaho.di.trans.steps.validator;

import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.pentaho.di.core.Const;
import org.pentaho.di.core.RowSet;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettlePluginException;
import org.pentaho.di.core.exception.KettleStepException;
import org.pentaho.di.core.exception.KettleValueException;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.core.row.value.ValueMetaFactory;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaDataCombi;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.trans.step.errorhandling.StreamInterface;

/**
 * Calculate new field values using pre-defined functions.
 *
 * @author Matt
 * @since 8-sep-2005
 */
public class Validator extends BaseStep implements StepInterface {
  private static Class<?> PKG = ValidatorMeta.class; // for i18n purposes, needed by Translator2!!

  public class FieldIndexes {
    public int indexName;
    public int indexA;
    public int indexB;
    public int indexC;
  }

  private ValidatorMeta meta;
  private ValidatorData data;

  public Validator( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
    Trans trans ) {
    super( stepMeta, stepDataInterface, copyNr, transMeta, trans );
  }

  public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException {
    meta = (ValidatorMeta) smi;
    data = (ValidatorData) sdi;

    Object[] r;

    if ( first ) {
      first = false;

      readSourceValuesFromInfoSteps();

      // Read the row AFTER the info rows
      // That way the info-rowsets are out of the way
      //
      r = getRow(); // get row, set busy!
      if ( r == null ) { // no more input to be expected...

        setOutputDone();
        return false;
      }

      data.fieldIndexes = new int[meta.getValidations().size()];

      // Calculate the indexes of the values and arguments in the target data or temporary data
      // We do this in advance to save time later on.
      //
      for ( int i = 0; i < meta.getValidations().size(); i++ ) {
        Validation field = meta.getValidations().get( i );

        if ( !Const.isEmpty( field.getFieldName() ) ) {
          data.fieldIndexes[i] = getInputRowMeta().indexOfValue( field.getFieldName() );
          if ( data.fieldIndexes[i] < 0 ) {
            // Nope: throw an exception
            throw new KettleStepException( "Unable to find the specified fieldname '"
              + field.getFieldName() + "' for validation#" + ( i + 1 ) );
          }
        } else {
          throw new KettleStepException( "There is no name specified for validator field #" + ( i + 1 ) );
        }
      }
    } else {
      // Read the row AFTER the info rows
      // That way the info-rowsets are out of the way
      //
      r = getRow(); // get row, set busy!
      if ( r == null ) { // no more input to be expected...

        setOutputDone();
        return false;
      }
    }

    if ( log.isRowLevel() ) {
      logRowlevel( "Read row #" + getLinesRead() + " : " + getInputRowMeta().getString( r ) );
    }

    try {
      List<KettleValidatorException> exceptions = validateFields( getInputRowMeta(), r );
      if ( exceptions.size() > 0 ) {
        if ( getStepMeta().isDoingErrorHandling() ) {
          if ( meta.isConcatenatingErrors() ) {
            StringBuilder messages = new StringBuilder();
            StringBuilder fields = new StringBuilder();
            StringBuilder codes = new StringBuilder();
            boolean notFirst = false;
            for ( KettleValidatorException e : exceptions ) {
              if ( notFirst ) {
                messages.append( meta.getConcatenationSeparator() );
                fields.append( meta.getConcatenationSeparator() );
                codes.append( meta.getConcatenationSeparator() );
              } else {
                notFirst = true;
              }
              messages.append( e.getMessage() );
              fields.append( e.getFieldname() );
              codes.append( e.getCodeDesc() );
            }
            putError( getInputRowMeta(), r, exceptions.size(), messages.toString(), fields.toString(), codes
              .toString() );
          } else {
            for ( KettleValidatorException e : exceptions ) {
              putError( getInputRowMeta(), r, 1, e.getMessage(), e.getFieldname(), e.getCodeDesc() );
            }
          }
        } else {
          KettleValidatorException e = exceptions.get( 0 );
          throw new KettleException( e.getMessage(), e );
        }
      } else {
        putRow( getInputRowMeta(), r ); // copy row to possible alternate rowset(s).
      }
    } catch ( KettleValidatorException e ) {
      if ( getStepMeta().isDoingErrorHandling() ) {
        putError( getInputRowMeta(), r, 1, e.getMessage(), e.getFieldname(), e.getCodeDesc() );
      } else {
        throw new KettleException( e.getMessage(), e );
      }
    }

    if ( log.isRowLevel() ) {
      logRowlevel( "Wrote row #" + getLinesWritten() + " : " + getInputRowMeta().getString( r ) );
    }
    if ( checkFeedback( getLinesRead() ) ) {
      logBasic( "Linenr " + getLinesRead() );
    }

    return true;
  }

  private void readSourceValuesFromInfoSteps() throws KettleStepException {
    for ( int i = 0; i < meta.getValidations().size(); i++ ) {
      Validation field = meta.getValidations().get( i );
      List<StreamInterface> streams = meta.getStepIOMeta().getInfoStreams();

      // If we need to source the allowed values data from a different step, we do this here as well
      //
      if ( field.isSourcingValues() ) {
        if ( streams.get( i ).getStepMeta() == null ) {
          throw new KettleStepException(
            "There is no valid source step specified for the allowed values of validation ["
              + field.getName() + "]" );
        }
        if ( Const.isEmpty( field.getSourcingField() ) ) {
          throw new KettleStepException(
            "There is no valid source field specified for the allowed values of validation ["
              + field.getName() + "]" );
        }

        // Still here : OK, read the data from the specified step...
        // The data is stored in data.listValues[i] and data.constantsMeta
        //
        RowSet allowedRowSet = findInputRowSet( streams.get( i ).getStepname() );
        int fieldIndex = -1;
        List<Object> allowedValues = new ArrayList<Object>();
        Object[] allowedRowData = getRowFrom( allowedRowSet );
        while ( allowedRowData != null ) {
          RowMetaInterface allowedRowMeta = allowedRowSet.getRowMeta();
          if ( fieldIndex < 0 ) {
            fieldIndex = allowedRowMeta.indexOfValue( field.getSourcingField() );
            if ( fieldIndex < 0 ) {
              throw new KettleStepException( "Source field ["
                + field.getSourcingField() + "] is not found in the source row data" );
            }
            data.constantsMeta[i] = allowedRowMeta.getValueMeta( fieldIndex );
          }
          Object allowedValue = allowedRowData[fieldIndex];
          if ( allowedValue != null ) {
            allowedValues.add( allowedValue );
          }

          // Grab another row too...
          //
          allowedRowData = getRowFrom( allowedRowSet );
        }
        // Set the list values in the data block...
        //
        data.listValues[i] = allowedValues.toArray( new Object[allowedValues.size()] );
      }
    }
  }

  /**
   * @param inputRowMeta the input row metadata
   * @param r            the input row (data)
   * @throws KettleValidatorException in case there is a validation error, details are stored in the exception.
   */
  private List<KettleValidatorException> validateFields( RowMetaInterface inputRowMeta, Object[] r ) throws KettleValueException {
    List<KettleValidatorException> exceptions = new ArrayList<KettleValidatorException>();

    for ( int i = 0; i < meta.getValidations().size(); i++ ) {
      Validation field = meta.getValidations().get( i );

      int valueIndex = data.fieldIndexes[i];
      ValueMetaInterface validatorMeta = data.constantsMeta[i];

      ValueMetaInterface valueMeta = inputRowMeta.getValueMeta( valueIndex );
      Object valueData = r[valueIndex];

      // Check for null
      //
      boolean isNull = valueMeta.isNull( valueData );
      if ( !field.isNullAllowed() && isNull ) {
        KettleValidatorException exception =
          new KettleValidatorException(
            this,
            field,
            KettleValidatorException.ERROR_NULL_VALUE_NOT_ALLOWED,
            BaseMessages.getString(
              PKG, "Validator.Exception.NullNotAllowed", field.getFieldName(), inputRowMeta.getString( r ) ),
            field.getFieldName() );
        exceptions.add( exception );
        if ( !meta.isValidatingAll() ) {
          return exceptions;
        }
      }

      if ( field.isOnlyNullAllowed() && !isNull ) {
        KettleValidatorException exception =
          new KettleValidatorException(
            this,
            field,
            KettleValidatorException.ERROR_ONLY_NULL_VALUE_ALLOWED,
            BaseMessages.getString(
              PKG, "Validator.Exception.OnlyNullAllowed", field.getFieldName(), inputRowMeta.getString( r ) ),
            field.getFieldName() );
        exceptions.add( exception );
        if ( !meta.isValidatingAll() ) {
          return exceptions;
        }
      }

      // Check the data type!
      //
      if ( field.isDataTypeVerified() && field.getDataType() != ValueMetaInterface.TYPE_NONE ) {

        // Same data type?
        //
        if ( field.getDataType() != valueMeta.getType() ) {
          KettleValidatorException exception =
            new KettleValidatorException(
              this, field, KettleValidatorException.ERROR_UNEXPECTED_DATA_TYPE, BaseMessages.getString(
                PKG, "Validator.Exception.UnexpectedDataType", field.getFieldName(), valueMeta
                  .toStringMeta(), validatorMeta.toStringMeta() ), field.getFieldName() );
          exceptions.add( exception );
          if ( !meta.isValidatingAll() ) {
            return exceptions;
          }
        }
      }

      // Check various things if the value is not null..
      //
      if ( !isNull ) {

        if ( data.fieldsMinimumLengthAsInt[i] >= 0
          || data.fieldsMaximumLengthAsInt[i] >= 0 || data.minimumValue[i] != null
          || data.maximumValue[i] != null || data.listValues[i].length > 0 || field.isSourcingValues()
          || !Const.isEmpty( data.startString[i] ) || !Const.isEmpty( data.endString[i] )
          || !Const.isEmpty( data.startStringNotAllowed[i] ) || !Const.isEmpty( data.endStringNotAllowed[i] )
          || field.isOnlyNumericAllowed() || data.patternExpected[i] != null
          || data.patternDisallowed[i] != null ) {

          String stringValue = valueMeta.getString( valueData );
          int stringLength = stringValue.length();

          // Minimum length
          //
          // if (field.getMinimumLength()>=0 && stringValue.length()<field.getMinimumLength() ) {
          if ( data.fieldsMinimumLengthAsInt[i] >= 0 && stringLength < data.fieldsMinimumLengthAsInt[i] ) {
            KettleValidatorException exception =
              new KettleValidatorException(
                this, field, KettleValidatorException.ERROR_SHORTER_THAN_MINIMUM_LENGTH, BaseMessages
                  .getString(
                    PKG, "Validator.Exception.ShorterThanMininumLength", field.getFieldName(), valueMeta
                      .getString( valueData ), Integer.toString( stringValue.length() ), field
                      .getMinimumLength() ), field.getFieldName() );
            exceptions.add( exception );
            if ( !meta.isValidatingAll() ) {
              return exceptions;
            }
          }

          // Maximum length
          //
          // if (field.getMaximumLength()>=0 && stringValue.length()>field.getMaximumLength() ) {
          if ( data.fieldsMaximumLengthAsInt[i] >= 0 && stringLength > data.fieldsMaximumLengthAsInt[i] ) {
            KettleValidatorException exception =
              new KettleValidatorException(
                this, field, KettleValidatorException.ERROR_LONGER_THAN_MAXIMUM_LENGTH, BaseMessages
                  .getString(
                    PKG, "Validator.Exception.LongerThanMaximumLength", field.getFieldName(), valueMeta
                      .getString( valueData ), Integer.toString( stringValue.length() ), field
                      .getMaximumLength() ), field.getFieldName() );
            exceptions.add( exception );
            if ( !meta.isValidatingAll() ) {
              return exceptions;
            }
          }

          // Minimal value
          //
          if ( data.minimumValue[i] != null
            && valueMeta.compare( valueData, validatorMeta, data.minimumValue[i] ) < 0 ) {
            KettleValidatorException exception =
              new KettleValidatorException(
                this, field, KettleValidatorException.ERROR_LOWER_THAN_ALLOWED_MINIMUM,
                BaseMessages.getString(
                  PKG, "Validator.Exception.LowerThanMinimumValue", field.getFieldName(), valueMeta
                    .getString( valueData ), data.constantsMeta[i].getString( data.minimumValue[i] ) ),
                field.getFieldName() );
            exceptions.add( exception );
            if ( !meta.isValidatingAll() ) {
              return exceptions;
            }
          }

          // Maximum value
          //
          if ( data.maximumValue[i] != null
            && valueMeta.compare( valueData, validatorMeta, data.maximumValue[i] ) > 0 ) {
            KettleValidatorException exception =
              new KettleValidatorException(
                this, field, KettleValidatorException.ERROR_HIGHER_THAN_ALLOWED_MAXIMUM,
                BaseMessages.getString( PKG, "Validator.Exception.HigherThanMaximumValue", field
                  .getFieldName(), valueMeta.getString( valueData ), data.constantsMeta[i]
                  .getString( data.maximumValue[i] ) ), field.getFieldName() );
            exceptions.add( exception );
            if ( !meta.isValidatingAll() ) {
              return exceptions;
            }
          }

          // In list?
          //
          if ( field.isSourcingValues() || data.listValues[i].length > 0 ) {
            boolean found = false;
            for ( Object object : data.listValues[i] ) {
              if ( object != null
                && data.listValues[i] != null && valueMeta.compare( valueData, validatorMeta, object ) == 0 ) {
                found = true;
              }
            }
            if ( !found ) {
              KettleValidatorException exception =
                new KettleValidatorException(
                  this, field, KettleValidatorException.ERROR_VALUE_NOT_IN_LIST, BaseMessages.getString(
                    PKG, "Validator.Exception.NotInList", field.getFieldName(), valueMeta
                      .getString( valueData ) ), field.getFieldName() );
              exceptions.add( exception );
              if ( !meta.isValidatingAll() ) {
                return exceptions;
              }
            }
          }

          // Numeric data or strings with only
          if ( field.isOnlyNumericAllowed() ) {
            KettleValidatorException exception = assertNumeric( valueMeta, valueData, field );
            if ( exception != null ) {
              exceptions.add( exception );
              if ( !meta.isValidatingAll() ) {
                return exceptions;
              }
            }
          }

          // Does not start with string value
          //
          if ( !Const.isEmpty( data.startString[i] ) && !stringValue.startsWith( data.startString[i] ) ) {
            KettleValidatorException exception =
              new KettleValidatorException(
                this, field, KettleValidatorException.ERROR_DOES_NOT_START_WITH_STRING, BaseMessages
                  .getString(
                    PKG, "Validator.Exception.DoesNotStartWithString", field.getFieldName(), valueMeta
                      .getString( valueData ), field.getStartString() ), field.getFieldName() );
            exceptions.add( exception );
            if ( !meta.isValidatingAll() ) {
              return exceptions;
            }
          }

          // Ends with string value
          //
          if ( !Const.isEmpty( data.endString[i] ) && !stringValue.endsWith( data.endString[i] ) ) {
            KettleValidatorException exception =
              new KettleValidatorException(
                this, field, KettleValidatorException.ERROR_DOES_NOT_END_WITH_STRING, BaseMessages.getString(
                  PKG, "Validator.Exception.DoesNotStartWithString", field.getFieldName(), valueMeta
                    .getString( valueData ), field.getEndString() ), field.getFieldName() );
            exceptions.add( exception );
            if ( !meta.isValidatingAll() ) {
              return exceptions;
            }
          }

          // Starts with string value
          //
          if ( !Const.isEmpty( data.startStringNotAllowed[i] )
            && stringValue.startsWith( data.startStringNotAllowed[i] ) ) {
            KettleValidatorException exception =
              new KettleValidatorException(
                this, field, KettleValidatorException.ERROR_STARTS_WITH_STRING, BaseMessages.getString(
                  PKG, "Validator.Exception.StartsWithString", field.getFieldName(), valueMeta
                    .getString( valueData ), field.getStartStringNotAllowed() ), field.getFieldName() );
            exceptions.add( exception );
            if ( !meta.isValidatingAll() ) {
              return exceptions;
            }
          }

          // Ends with string value
          //
          if ( !Const.isEmpty( data.endStringNotAllowed[i] ) && stringValue.endsWith( data.endStringNotAllowed[i] ) ) {
            KettleValidatorException exception =
              new KettleValidatorException(
                this, field, KettleValidatorException.ERROR_ENDS_WITH_STRING, BaseMessages.getString(
                  PKG, "Validator.Exception.EndsWithString", field.getFieldName(), valueMeta
                    .getString( valueData ), field.getEndStringNotAllowed() ), field.getFieldName() );
            exceptions.add( exception );
            if ( !meta.isValidatingAll() ) {
              return exceptions;
            }
          }

          // Matching regular expression allowed?
          //
          if ( data.patternExpected[i] != null ) {
            Matcher matcher = data.patternExpected[i].matcher( stringValue );
            if ( !matcher.matches() ) {
              KettleValidatorException exception =
                new KettleValidatorException(
                  this, field, KettleValidatorException.ERROR_MATCHING_REGULAR_EXPRESSION_EXPECTED,
                  BaseMessages.getString( PKG, "Validator.Exception.MatchingRegExpExpected", field
                    .getFieldName(), valueMeta.getString( valueData ), data.regularExpression[i] ), field
                    .getFieldName() );
              exceptions.add( exception );
              if ( !meta.isValidatingAll() ) {
                return exceptions;
              }
            }
          }

          // Matching regular expression NOT allowed?
          //
          if ( data.patternDisallowed[i] != null ) {
            Matcher matcher = data.patternDisallowed[i].matcher( stringValue );
            if ( matcher.matches() ) {
              KettleValidatorException exception =
                new KettleValidatorException(
                  this,
                  field,
                  KettleValidatorException.ERROR_MATCHING_REGULAR_EXPRESSION_NOT_ALLOWED,
                  BaseMessages.getString( PKG, "Validator.Exception.MatchingRegExpNotAllowed", field
                    .getFieldName(), valueMeta.getString( valueData ), data.regularExpressionNotAllowed[i] ),
                  field.getFieldName() );
              exceptions.add( exception );
              if ( !meta.isValidatingAll() ) {
                return exceptions;
              }
            }
          }
        }
      }
    }

    return exceptions;
  }

  // package-local visibility for testing purposes
  KettleValidatorException assertNumeric( ValueMetaInterface valueMeta,
                                          Object valueData,
                                          Validation field ) throws KettleValueException {
    if ( valueMeta.isNumeric() || containsOnlyDigits( valueMeta.getString( valueData ) ) ) {
      return null;
    }
    return new KettleValidatorException( this, field, KettleValidatorException.ERROR_NON_NUMERIC_DATA,
      BaseMessages.getString( PKG, "Validator.Exception.NonNumericDataNotAllowed", field.getFieldName(),
        valueMeta.toStringMeta() ), field.getFieldName() );
  }

  private boolean containsOnlyDigits( String string ) {
    for ( char c : string.toCharArray() ) {
      if ( c < '0' || c > '9' ) {
        return false;
      }
    }
    return true;
  }

  public boolean init( StepMetaInterface smi, StepDataInterface sdi ) {
    meta = (ValidatorMeta) smi;
    data = (ValidatorData) sdi;

    if ( super.init( smi, sdi ) ) {
      // initialize steps by names
      List<StepMeta> steps = new ArrayList<>();
      for ( StepMetaDataCombi s : getTrans().getSteps() ) {
        steps.add( s.stepMeta );
      }
      meta.searchInfoAndTargetSteps( steps );

      // initialize arrays of validation data
      data.constantsMeta = new ValueMetaInterface[meta.getValidations().size()];
      data.minimumValueAsString = new String[meta.getValidations().size()];
      data.maximumValueAsString = new String[meta.getValidations().size()];
      data.fieldsMinimumLengthAsInt = new int[meta.getValidations().size()];
      data.fieldsMaximumLengthAsInt = new int[meta.getValidations().size()];
      data.minimumValue = new Object[meta.getValidations().size()];
      data.maximumValue = new Object[meta.getValidations().size()];
      data.listValues = new Object[meta.getValidations().size()][];
      data.errorCode = new String[meta.getValidations().size()];
      data.errorDescription = new String[meta.getValidations().size()];
      data.conversionMask = new String[meta.getValidations().size()];
      data.decimalSymbol = new String[meta.getValidations().size()];
      data.groupingSymbol = new String[meta.getValidations().size()];
      data.maximumLength = new String[meta.getValidations().size()];
      data.minimumLength = new String[meta.getValidations().size()];
      data.startString = new String[meta.getValidations().size()];
      data.endString = new String[meta.getValidations().size()];
      data.startStringNotAllowed = new String[meta.getValidations().size()];
      data.endStringNotAllowed = new String[meta.getValidations().size()];
      data.regularExpression = new String[meta.getValidations().size()];
      data.regularExpressionNotAllowed = new String[meta.getValidations().size()];
      data.patternExpected = new Pattern[meta.getValidations().size()];
      data.patternDisallowed = new Pattern[meta.getValidations().size()];

      for ( int i = 0; i < meta.getValidations().size(); i++ ) {

        Validation field = meta.getValidations().get( i );
        try {
          data.constantsMeta[i] = createValueMeta( field.getFieldName(), field.getDataType() );
          data.constantsMeta[i].setConversionMask( field.getConversionMask() );
          data.constantsMeta[i].setDecimalSymbol( field.getDecimalSymbol() );
          data.constantsMeta[i].setGroupingSymbol( field.getGroupingSymbol() );
          data.errorCode[i] = environmentSubstitute( Const.NVL( field.getErrorCode(), "" ) );
          data.errorDescription[i] = environmentSubstitute( Const.NVL( field.getErrorDescription(), "" ) );
          data.conversionMask[i] = environmentSubstitute( Const.NVL( field.getConversionMask(), "" ) );
          data.decimalSymbol[i] = environmentSubstitute( Const.NVL( field.getDecimalSymbol(), "" ) );
          data.groupingSymbol[i] = environmentSubstitute( Const.NVL( field.getGroupingSymbol(), "" ) );
          data.maximumLength[i] = environmentSubstitute( Const.NVL( field.getMaximumLength(), "" ) );
          data.minimumLength[i] = environmentSubstitute( Const.NVL( field.getMinimumLength(), "" ) );
          data.maximumValueAsString[i] = environmentSubstitute( Const.NVL( field.getMaximumValue(), "" ) );
          data.minimumValueAsString[i] = environmentSubstitute( Const.NVL( field.getMinimumValue(), "" ) );
          data.startString[i] = environmentSubstitute( Const.NVL( field.getStartString(), "" ) );
          data.endString[i] = environmentSubstitute( Const.NVL( field.getEndString(), "" ) );
          data.startStringNotAllowed[i] =
            environmentSubstitute( Const.NVL( field.getStartStringNotAllowed(), "" ) );
          data.endStringNotAllowed[i] = environmentSubstitute( Const.NVL( field.getEndStringNotAllowed(), "" ) );
          data.regularExpression[i] = environmentSubstitute( Const.NVL( field.getRegularExpression(), "" ) );
          data.regularExpressionNotAllowed[i] =
            environmentSubstitute( Const.NVL( field.getRegularExpressionNotAllowed(), "" ) );

          ValueMetaInterface stringMeta = cloneValueMeta( data.constantsMeta[i], ValueMetaInterface.TYPE_STRING );
          data.minimumValue[i] =
            Const.isEmpty( data.minimumValueAsString[i] ) ? null : data.constantsMeta[i].convertData(
              stringMeta, data.minimumValueAsString[i] );
          data.maximumValue[i] =
            Const.isEmpty( data.maximumValueAsString[i] ) ? null : data.constantsMeta[i].convertData(
              stringMeta, data.maximumValueAsString[i] );

          try {
            data.fieldsMinimumLengthAsInt[i] = Integer.valueOf( Const.NVL( data.minimumLength[i], "-1" ) );
          } catch ( NumberFormatException nfe ) {
            throw new KettleValueException(
              "Caught a number format exception converting minimum length with value "
                + data.minimumLength[i] + " to an int.", nfe );
          }

          try {
            data.fieldsMaximumLengthAsInt[i] = Integer.valueOf( Const.NVL( data.maximumLength[i], "-1" ) );
          } catch ( NumberFormatException nfe ) {
            throw new KettleValueException(
              "Caught a number format exception converting minimum length with value "
                + data.maximumLength[i] + " to an int.", nfe );
          }

          int listSize = field.getAllowedValues() != null ? field.getAllowedValues().length : 0;
          data.listValues[i] = new Object[listSize];
          for ( int s = 0; s < listSize; s++ ) {
            data.listValues[i][s] =
              Const.isEmpty( field.getAllowedValues()[s] ) ? null : data.constantsMeta[i].convertData(
                stringMeta, environmentSubstitute( field.getAllowedValues()[s] ) );
          }
        } catch ( KettleException e ) {
          if ( field.getDataType() == ValueMetaInterface.TYPE_NONE ) {
            logError( BaseMessages.getString( PKG, "Validator.Exception.SpecifyDataType" ), e );
          } else {
            logError( BaseMessages.getString( PKG, "Validator.Exception.DataConversionErrorEncountered" ), e );
          }
          return false;
        }

        if ( !Const.isEmpty( data.regularExpression[i] ) ) {
          data.patternExpected[i] = Pattern.compile( data.regularExpression[i] );
        }
        if ( !Const.isEmpty( data.regularExpressionNotAllowed[i] ) ) {
          data.patternDisallowed[i] = Pattern.compile( data.regularExpressionNotAllowed[i] );
        }

      }

      return true;
    }
    return false;
  }

  protected ValueMetaInterface createValueMeta( String name, int type ) throws KettlePluginException {
    return ValueMetaFactory.createValueMeta( name, type );
  }

  protected ValueMetaInterface cloneValueMeta( ValueMetaInterface valueMeta, int type ) throws KettlePluginException {
    return ValueMetaFactory.cloneValueMeta( valueMeta, type );
  }

}
